package com.shinewind.common.util;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.text.csv.*;
import cn.hutool.core.util.CharsetUtil;
import cn.hutool.core.util.StrUtil;
import com.shinewind.pojo.entity.Rating;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.ml.evaluation.RegressionEvaluator;
import org.apache.spark.ml.recommendation.ALS;
import org.apache.spark.ml.recommendation.ALSModel;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.io.*;
import java.util.*;
import java.util.stream.Collectors;

public class RecommendUtils {
    //创建或覆盖data.csv文件,用于提供训练模型所需的数据
    public static void createDataCsv(List<Rating> ratings) throws IOException {
        CsvWriter csvWriter;
        List<Rating> ratingsList = ratings;
        List<String[]> stringList = ratingsList.stream().map(rating1 -> {
            String[] strings = new String[3];
            strings[0] = String.valueOf(rating1.getUserId());
            strings[1] = String.valueOf(rating1.getModelId());
            strings[2] = String.valueOf(rating1.getRating());
            return strings;
        }).collect(Collectors.toList());
        //获取当前resource目录路径
        File file = new File("");
        String filePath = file.getCanonicalFile() + File.separator + "model_player_task\\src\\main\\resources\\data.csv";
        System.out.println("filePath:"+filePath);
        //生成并写入csv文件
        csvWriter = CsvUtil.getWriter(filePath, CharsetUtil.CHARSET_GBK);
        csvWriter.writeHeaderLine("userId","itemId","rating");
        System.out.println("stringList:"+stringList);
        csvWriter.write(stringList);
    }
    //训练模型,基于ALS协同推荐算法给用户推荐n个商品
    public static void createTestCsv(int n) throws IOException {
        File file = new File("");
        String filePath = file.getCanonicalFile() + File.separator + "model_player_task\\src\\main\\resources\\data.csv";
        //连接spark
        SparkSession spark = SparkSession.builder().master("local").appName("model_player").getOrCreate();
        //读取生成的csv文件,转换成JavaRDD类型
        JavaRDD<String> ratingRDD = spark.read().textFile(filePath).javaRDD();
        String headerLine = ratingRDD.first();
        ratingRDD = ratingRDD.filter(line ->{
            return !line.equals(headerLine);
        });
        JavaRDD<Rating> ratingJavaRDD = ratingRDD.map(Rating::parseRating);
        //将JavaRDD转换成dataFrame格式,用于拆分训练
        Dataset<Row> ratings = spark.createDataFrame(ratingJavaRDD,Rating.class);
        //将数据切分成两份
        Dataset<Row>[] splits = ratings.randomSplit(new double[]{0.8,0.2});
        Dataset<Row> trainingData = splits[0];
        Dataset<Row> testData = splits[1];
        /**
         * @see maxIter 要运行的最大迭代次数
         * @see rank 模型中潜在因素的数值
         * @see regParam 指定的正则化参数
         */
        // 对训练数据集使用ALS算法,并构建模型
        ALS als = new ALS().setMaxIter(10).setRank(5).setRegParam(0.01).setUserCol("userId")
                .setItemCol("modelId").setRatingCol("rating");
        ALSModel model = als.fit(trainingData);
        //冷启动,去除NAN指标
        model.setColdStartStrategy("drop");
        // 通过计算均方根误差rmse(Root Mean Squared Error)对测试数据集评估模型
        Dataset<Row> predictions = model.transform(testData);
        //计算均方根误差,预测值与真实值的偏差的平方除以观测次数,开根号
        RegressionEvaluator evaluator = new RegressionEvaluator().setMetricName("rmse")
                .setLabelCol("rating").setPredictionCol("prediction");
        if(predictions.count() > 0){
            double rmse = evaluator.evaluate(predictions);
            System.out.println("均方根误差rmse:"+rmse);
            //进行推荐
            Dataset<Row> modelRecs = model.recommendForAllUsers(n);
            List<Object []> resultList = modelRecs.collectAsList().stream().map(model1 ->{
                Object[] objects = new Object[2];
                objects[0] = model1.get(0).toString();
                //TODO格式化推荐的模型id
                objects[1] = model1.getList(1).toArray();
                return objects;
            }).collect(Collectors.toList());
            //生成并写入csv文件
            CsvWriter csvWriter;
            String outPath = file.getCanonicalFile() + File.separator + "model_player_task\\src\\main\\resources\\test.csv";
            csvWriter = CsvUtil.getWriter(outPath, CharsetUtil.CHARSET_GBK);
            csvWriter.writeHeaderLine("userId","recommendation");
            System.out.println("stringList:"+resultList);
            csvWriter.write(resultList);
        }
    }
    //读取testCsv文件
    public static Map<Integer,List<Integer>> readTestCsv() throws IOException {
        File file = new File("");
        String filePath = file.getCanonicalFile() + File.separator + "model_player_task\\src\\main\\resources\\test.csv";
        CsvReader csvReader = CsvUtil.getReader();
        CsvData data= csvReader.read(FileUtil.file(filePath),CharsetUtil.CHARSET_GBK);
        List<CsvRow> csvRows = data.getRows();
        csvRows = ListUtil.sub(csvRows,1,csvRows.size());
        Map<Integer,List<Integer>> recommendMap = new HashMap<>();
        for(CsvRow row:csvRows){
            System.out.println("row:"+row.getRawList());
            int userId = Integer.parseInt(row.getRawList().get(0));
            String[] recommendation = row.getRawList().get(1).split(",");
            List<Integer> modelIds = new ArrayList<>();
            for(int i=0;i<recommendation.length;i++){
                if(recommendation[i].indexOf("[[")!=-1 || recommendation[i].indexOf("[")!=-1){
                    recommendation[i] = StrUtil.replace(recommendation[i],"[[","").trim();
                    recommendation[i] = StrUtil.replace(recommendation[i],"[","").trim();
                    modelIds.add(Integer.parseInt(recommendation[i]));
                }
            }
            recommendMap.put(userId,modelIds);
        }
        return recommendMap;
    }
    //判断文件是否存在
    public static Boolean testCsvIsExist() throws IOException {
        File file = new File("");
        String outPath = file.getCanonicalFile() + File.separator + "model_player_task\\src\\main\\resources\\test.csv";
        if(FileUtil.exist(outPath)==true)
            return true;
        else
            return false;
    }
    //判断文件是否存在
    public static Boolean dataCsvIsExist() throws IOException {
        File file = new File("");
        String filePath = file.getCanonicalFile() + File.separator + "model_player_task\\src\\main\\resources\\data.csv";
        if(FileUtil.exist(filePath)==true)
            return true;
        else
            return false;
    }

    public static void main(String[] args) throws IOException {
        /*List<Rating> ratings = new ArrayList<>();
        ratings.add(new Rating(1,1,4.0));
        ratings.add(new Rating(1,2,3.0));
        ratings.add(new Rating(2,1,4.0));
        ratings.add(new Rating(3,1,2.0));
        ratings.add(new Rating(2,3,1.0));
        createDataCsv(ratings);*/
        //createTestCsv(10);
        /*Map<Integer,List<Integer>> recommendMap = readTestCsv();
        Iterator<Map.Entry<Integer,List<Integer>>> entries = recommendMap.entrySet().iterator();
        while(entries.hasNext()){
            Map.Entry<Integer,List<Integer>> entry = entries.next();
            System.out.println("userId:"+entry.getKey()+" modelIds:"+entry.getValue());
        }*/
    }
}
